package test.mllib.correlation;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.mllib.linalg.Matrix;
import org.apache.spark.mllib.linalg.Vector;
import org.apache.spark.mllib.linalg.Vectors;
import org.apache.spark.mllib.stat.Statistics;

public class PearsonTest3 {

	public static void main(String[] args) {
		SparkConf sparkConf = new SparkConf().setAppName("LinearRegressionTest").setMaster("local[1]");
		JavaSparkContext sc = new JavaSparkContext(sparkConf);
		JavaRDD<String> data = sc.textFile("sparkdata/mllib/abalone.data");

		JavaRDD<Vector> parsedData = data.map(new Function<String, Vector>() {
	        private static final long serialVersionUID = 1L;
			@Override
			public Vector call(String line) throws Exception {
			    String[] xnStr = line.split(",");
				double[] xn = new double[xnStr.length];
				if("M".equals(xnStr[0])){
					xn[0] = 0;
				}
				if("F".equals(xnStr[0])){
					xn[0] = 1;
				}
				if("I".equals(xnStr[0])){
					xn[0] = 2;
				}
				for (int i = 1; i < xn.length; i++) {
					xn[i] = Double.parseDouble(xnStr[i]);
				}
				return Vectors.dense(xn);
			}
		}).cache();
		// spearman | pearson
		Matrix correlMatrix = Statistics.corr(parsedData.rdd(), "pearson");
		for(int i = 0; i < 8; i++){
			for(int j = 0; j < 8; j++){
				System.out.print(correlMatrix.apply(i, j));
				System.out.print(" ");
			}
			System.out.println();
		}
		
	}
}
